package com.etl.dataflow.executor.core;

import com.etl.dataflow.common.exception.EtlException;
import com.etl.dataflow.executor.element.Row;
import com.etl.dataflow.executor.element.TerminateRow;

import java.util.List;

/**
 * @author dx
 * @since 2022/4/14
 */
public class Transporter implements ChannelReader, ChannelWriter {

    private final MemoryChannel<Row> channel;

    private boolean isShutdown = false;

    public Transporter(MemoryChannel<Row> channel) {
        this.channel = channel;
    }

    @Override
    public Row readOneRow() {
        if (isShutdown) {
            throw new EtlException("无法从已经shutdown的transport中读取元素");
        }
        // 读不出则阻塞
        Row row = channel.take();
        if (row instanceof TerminateRow) {
            return null;
        }
        return row;
    }

    @Override
    public void writeOneRow(Row row) {
        if (isShutdown) {
            throw new EtlException("无法向已经shutdown的transport中写元素");
        }
        // 写满则阻塞
        this.channel.add(row);
    }

    @Override
    public void terminate() {
        this.channel.add(new TerminateRow());
    }

    @Override
    public void shutdown() {
        this.isShutdown = true;
        try {
            channel.clear();
        } catch (Throwable t) {
            t.printStackTrace();
        }

    }
}
